package com.wz.utils.batch.aspectj;


import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import com.wz.utils.batch.annotation.BatchDealWithAnt;
import com.wz.utils.batch.exception.BaseException;
import com.wz.utils.batch.exception.BatchWithRollBack;
import com.wz.utils.batch.exception.MsgException;
import com.wz.utils.batch.service.BatchWithService;
import lombok.extern.slf4j.Slf4j;
import org.aspectj.lang.ProceedingJoinPoint;
import org.aspectj.lang.Signature;
import org.aspectj.lang.annotation.Around;
import org.aspectj.lang.annotation.Aspect;
import org.aspectj.lang.annotation.Pointcut;
import org.aspectj.lang.reflect.MethodSignature;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.lang.reflect.Method;
import java.sql.SQLException;
import java.util.List;
import java.util.concurrent.*;

@Aspect
@Component
@Slf4j
public class BatchDealWithAspectj {

    /**
     * 批量处理子任务接口
     *
     **/
    @Resource
    private BatchWithService batchWithService;


    /**
     * 批量处理拦截切入标识
     **/
    @Pointcut("@annotation(com.wz.utils.batch.annotation.BatchDealWithAnt)")
    public void batchDealWithPointCut(){

    }

    /**
     * @Description: 方法环绕
     * @Author: ChenGang
     * @Date: 2021/8/31 15:45
     **/
    @Around(value = "batchDealWithPointCut()")
    public Object batchDealWithAround(ProceedingJoinPoint joinPoint) throws BaseException, SQLException {
        //声明返回值
        Object res = null;
        //获取方法签名
        Signature signature = joinPoint.getSignature();
        MethodSignature methodSignature = (MethodSignature) signature;
        //获取方法对象
        Method method = null;
        try {
            method = joinPoint.getTarget().getClass().getMethod(joinPoint.getSignature().getName(), methodSignature.getParameterTypes());
        } catch (NoSuchMethodException e) {
            throw new BaseException("=======批量处理异常！");
        }
        String methodName = method.getName();
        String className = joinPoint.getTarget().getClass().getSimpleName();
        BatchDealWithAnt batchDealWithAnt = method.getAnnotation(BatchDealWithAnt.class);
        try {
            //执行处理逻辑
            res =  mainThreadRun(joinPoint,batchDealWithAnt,className,methodName);

        } catch (Throwable throwable) {
            throwable.printStackTrace();
            log.error("=======批量处理：{}--{}异常！",className,methodName);
            throw new BaseException("=======批量处理："+className + "--"+methodName+"异常！");
        }
        return res;
    }


    /**
     * @Description: 任务分配逻辑
     * @Author: ChenGang
     * @Date: 2021/8/31 15:40
     **/
    public Object mainThreadRun(ProceedingJoinPoint joinPoint, BatchDealWithAnt batchDealWithAnt, String className, String methodName) throws Throwable {
        Object res = null;
        //获取参数列表
        Object[] args = joinPoint.getArgs();
        Object tasks = args[batchDealWithAnt.index()];
        if (tasks instanceof List){
            List tasksList = (List) tasks;
            if (tasksList.size() < batchDealWithAnt.minMath()){
                log.info("=======批量处理：{}--{}低于处理阈值，正常执行......",className,methodName);
                res = joinPoint.proceed(args);
                return res;
            }
            if (batchDealWithAnt.timeout() > 30){
                log.info("=======批量处理：{}--{}超过最大等待时间，正常执行......",className,methodName);
                res = joinPoint.proceed(args);
                return res;
            }

            log.info("=======批量处理：为{}--{}高于处理阈值,预分配任务中......",className,methodName);

            //每条线程最小处理任务数
            int perThreadHandleCount = 1;
            //线程池的默认最大线程数
            int cupNum = Runtime.getRuntime().availableProcessors();
            int nThreads = (int)Math.floor(cupNum*0.7*2) ;
            int taskSize = tasksList.size();

            perThreadHandleCount = taskSize % nThreads == 0 ? taskSize / nThreads : taskSize / nThreads + 1;
            nThreads = taskSize % perThreadHandleCount == 0 ? taskSize / perThreadHandleCount : taskSize / perThreadHandleCount + 1;

            log.info("=======批量处理，共创建 {}个线程",nThreads);
            //监控主线程
            CountDownLatch mainLatch = new CountDownLatch(1);
            //监控子线程
            CountDownLatch threadLatch = new CountDownLatch(nThreads);
            //必须要使用对象，如果使用变量会造成线程之间不可共享变量值
            BatchWithRollBack rollBack = new BatchWithRollBack(false);
            //创建线程池
            ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("批量处理：" + className + " - " + methodName + "-pool-").build();
            ThreadPoolExecutor fixedThreadPool = new ThreadPoolExecutor(
                    nThreads,
                    nThreads,
                    batchDealWithAnt.timeout(),
                    TimeUnit.SECONDS,
                    new LinkedBlockingDeque<>(nThreads),
                    threadFactory,
                    new ThreadPoolExecutor.AbortPolicy());
            List<Future<Boolean>> futures = Lists.newArrayList();
            //根据子线程执行结果判断是否需要回滚
            BlockingDeque<Boolean> resultList = new LinkedBlockingDeque<>(nThreads);
            //给每个线程分配任务
            for (int i = 0; i < nThreads; i++) {
                int lastIndex = (i + 1) * perThreadHandleCount;
                List subList = tasksList.subList(i * perThreadHandleCount, lastIndex >= taskSize ? taskSize : lastIndex);
                //组装子任务参数
                Object[] params = new Object[args.length];
                for (int j = 0; j < args.length; j++) {
                    if (j == batchDealWithAnt.index()){
                        params[batchDealWithAnt.index()] = subList;
                        continue;
                    }
                    params[j] = args[j];
                }
                //创建子任务，丢给线程池处理
                BatchWithCallable threadHandleTaskCallable = new BatchWithCallable(mainLatch, threadLatch, rollBack,joinPoint,params,resultList);
                Future<Boolean> future = fixedThreadPool.submit(threadHandleTaskCallable);
                futures.add(future);
            }
            try {
                //等待所有子线程执行完毕
                boolean await = threadLatch.await(batchDealWithAnt.timeout(), TimeUnit.SECONDS);
                //如果超时，直接回滚
                if (!await) {
                    log.debug("==========批量处理线程执行超时");
                    rollBack.setRollBack(true);
                } else {
                    log.info("批量处理，线程执行完毕，共 {}个线程",nThreads);
                    //查看执行情况，如果有存在需要回滚的线程，则全部回滚
                    for (int i = 0; i < nThreads; i++) {
                        Boolean result = resultList.take();
                        log.debug("==========子线程返回结果result: "+result);
                        if (!result) {
                            /** 有线程执行异常，需要回滚子线程. */
                            rollBack.setRollBack(true);
                        }
                    }
                }
            } catch (InterruptedException e) {
                log.error("等待所有子线程执行完毕时，出现异常");
                rollBack.setRollBack(true);
                e.printStackTrace();
                throw new MsgException("等待所有子线程执行完毕时，出现异常，整体回滚");
            } finally {
                //子线程再次开始执行
                mainLatch.countDown();
                log.info("关闭线程池，释放资源");
                fixedThreadPool.shutdown();
            }
            if (rollBack.getRollBack()){
                throw new BaseException("有线程执行异常，主线程回滚");
            }
            log.info("======>批量处理成功");
        } else {
            log.error("批量处理：{}--{}参数异常,正常执行......",className,methodName);
            res = joinPoint.proceed(args);
        }
        return res;
    }


    /**
     * @Description: 批量处理多线程实现
     * @Author: ChenGang
     * @Date: 2021/8/31 15:25
     **/
    class BatchWithCallable implements Callable<Boolean> {

        /**
         * 主线程监控
         */
        private CountDownLatch mainLatch;
        /**
         * 子线程监控
         */
        private CountDownLatch threadLatch;
        /**
         * 是否回滚
         */
        private BatchWithRollBack rollBack;


        /**
         * 切点对象，通过此对象执行方法
         */
        private ProceedingJoinPoint joinPoint;


        /**
         * 方法执行参数
         */
        private Object[] args;

        /**
         *
         * 子线程执行结果判断
         **/
        private BlockingDeque<Boolean> resultList;



        public BatchWithCallable(CountDownLatch mainLatch, CountDownLatch threadLatch, BatchWithRollBack rollBack, ProceedingJoinPoint joinPoint, Object[] args, BlockingDeque<Boolean> resultList) {
            this.mainLatch = mainLatch;
            this.threadLatch = threadLatch;
            this.rollBack = rollBack;
            this.joinPoint = joinPoint;
            this.args = args;
            this.resultList = resultList;
        }

        @Override
        public Boolean call() throws Exception {
            try {
                return batchWithService.childThreadRun(joinPoint,args,threadLatch,mainLatch,rollBack,resultList);
            } catch (MsgException e) {
                e.printStackTrace();
            }
            return false;
        }
    }

}

